package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: kongcb
 * @Description: 变更记录表只有新增数据。
 * @Date: 2021/12/15 9:19
 */
public class TEciChange {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_eci_change").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_ent_alterations");
        long zs_count = zs_data.count();
        if(zs_count>0){

        //企查查变更表获取统一社会信用代码用于和中数变更表关联
//        spark.sql("SELECT a.*,\n" +
//                "       b.credit_code\n" +
//                "FROM a_dc_ep_dwi.t_eci_change a\n" +
//                "LEFT JOIN a_dc_ep_ods.t_eci_company b ON a.company_id = b.company_id\n" +
//                "WHERE b.company_id IS NOT NULL").createOrReplaceTempView("qcc_change_tmp");
            //中数变更表获取统一社会信用代码用于和企查查变更表关联
            spark.sql("SELECT a.*,\n" +
                    "       b.creditcode\n" +
                    "FROM a_dc_ep_ods.zs_t_ent_alterations a\n" +
                    "INNER JOIN a_dc_ep_ods.zs_t_ent_alterations_del b ON a.entid = b.entid  where b.creditcode is not null and b.creditcode !=''").createOrReplaceTempView("zs_change_tmp");
            //新增（inner join）
            spark.sql("insert overwrite table a_dc_ep_incr.t_eci_change select zs.record_id,\n" +
                    "qcc.key_no,\n" +
                    "qcc.company_id,\n" +
                    "qcc.company_name,\n" +
                    "zs.ALTITEM,\n" +
                    "zs.ALTBE,\n" +
                    "zs.ALTAF,\n" +
                    "zs.ALTDATE,\n" +
                    "zs.JOBID as dates,\n" +
                    "'1' as isadd from zs_change_tmp zs inner join a_dc_ep_ods.t_eci_company qcc on upper(trim(zs.creditcode)) = upper(trim(qcc.credit_code))");
//        //新增
//        spark.sql("insert into table a_dc_ep_incr.t_eci_change select " +
//                "zs_change.RECORD_ID,\n" +
//                " if(qcc_company.credit_code is not null,qcc_company.key_no,zs_change.key_no),\n" +
//                " if(qcc_company.credit_code is not null,qcc_company.company_id,zs_change.company_id),\n" +
//                " if(qcc_company.credit_code is not null,qcc_company.company_name,zs_change.company_name),\n" +
//                " zs_change.ALTITEM,\n" +
//                " zs_change.ALTBE,\n" +
//                " zs_change.ALTAF,\n" +
//                " zs_change.ALTDATE,\n" +
//                " zs_change.dates,\n" +
//                " zs_change.isadd " +
//                " from (select " +
//                "zs.RECORD_ID,\n" +
//                "md5(upper(trim(zs.creditcode))) as key_no,\n" +
//                "md5(upper(trim(zs.creditcode))) as company_id,\n" +
//                "'中数企业名称' as company_name,\n" +
//                "zs.ALTITEM,\n" +
//                "zs.ALTBE,\n" +
//                "zs.ALTAF,\n" +
//                "zs.ALTDATE,\n" +
//                "zs.JOBID as dates,\n" +
//                "'1' as isadd " +
//                "qcc.credit_code " +
//                "from zs_change_tmp zs left join qcc_change_tmp qcc on zs.creditcode = qcc.credit_code  where qcc.credit_code is null )  zs_change " +
//                "left join a_dc_ep_ods.t_eci_company qcc_company on zs_change.credit_code = qcc_company.credit_code");
        //删除
//        spark.sql("insert into table a_dc_ep_incr.t_eci_change select qcc.id,\n" +
//                "qcc.key_no,\n" +
//                "qcc.company_id,\n" +
//                "qcc.company_name,\n" +
//                "qcc.project_name,\n" +
//                "qcc.before_content,\n" +
//                "qcc.after_content,\n" +
//                "qcc.change_date,\n" +
//                "qcc.dates,\n" +
//                "'-1' as isadd from (select zs_del.* from a_dc_ep_ods.t_ent_alterations_del zs_del left join a_dc_ep_ods.t_ent_alterations a on zs_del.entid = a.entid where a.entid is null) zs inner join a_dc_ep_dwi.t_eci_change qcc on zs.creditcode = qcc.credit_code");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_eci_change");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
